package com.raja.requestForwarding.mq.rabbitmq.receive;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import com.raja.fleet.common.Constants;
import com.raja.fleet.mq.channel.CustomChannel;
import com.raja.fleet.service.CarService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 监听车辆数据是否有更新
 * Created by liuyanjun
 * on 2017/3/31.
 */
@Component
//@RabbitListener(queues = Constants.fanout_raja_fleet) //消息队列名称
public class CarListener implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(CarListener.class);

    @Autowired
    private CarService carService;
    @Autowired
    private CustomChannel customChannel;

//    @RabbitHandler
//    public void process(Object hello, Channel channel) {
//        DefaultConsumer consumer = new DefaultConsumer(channel);
//        boolean autoAck = false;//设置自动回馈；true  手动反馈 false
//        Message message = (Message) hello;
//        try {
//            channel.basicConsume(Constants.fanout_raja_fleet, autoAck,consumer);
//            String carInfo = new String(message.getBody(), "utf-8");
//            log.debug("Receiver : " + carInfo);
////            JSONObject jsonObject = JSONObject.parseObject(carInfo);
////            carService.updateCar(jsonObject);
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); //回复确认机制，消息队列才会删除消息； 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
//        }catch (Exception e){
//            log.error(e.getMessage(),e);
//            try {
//                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);//回复失败信息，消息队列不删除消息 ack返回false，并重新回到队列，api里面解释得很清楚
//            } catch (IOException e1) {
//                log.error(e1.getMessage(),e1);
//            }
//        }
//    }

    public void run(){
        GetResponse getResponse = null;
        Channel channel = customChannel.getChannel(Constants.fanout_raja_fleet);
        int count = 0;
        String carInfo= "";
        if(channel != null && channel.isOpen()) {
            log.info("车辆信息队列开启......");
            while (true) {
                try {
                    getResponse = channel.basicGet(Constants.fanout_raja_fleet, false);
                    if (getResponse != null) {
                         carInfo = new String(getResponse.getBody(), "utf-8");
                        log.info("接收到车辆的信息 : " + carInfo);
                          JSONObject jsonObject = JSONObject.parseObject(carInfo);
                         carService.updateCar(jsonObject);
                        channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);
                    } else {
                        //当没有消息的时候，暂停10秒钟
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    try {
                        //当发生异常，暂停20秒钟
                        Thread.sleep(20000);
                        if(getResponse != null) {
                            //如果消费20次，则抛弃，记录日志
                            if (count > 20) {
                                log.error("接收到车队账户的信息 : " + carInfo);
                                channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);
                                count = 0;
                            } else {
                                count++;
                                channel.basicNack(getResponse.getEnvelope().getDeliveryTag(), false, true);
                            }
                        }
                    } catch (Exception e1) {
                        log.error(e1.getMessage(), e1);
                    }
                }
            }
        }else {
            log.error("车辆mq通道不通");
        }
    }

}
